package com.cetccloud.ap.proxy.extractor;

import cn.hutool.core.bean.BeanUtil;
import com.cetccloud.ap.config.ProxyConfig;
import com.cetccloud.ap.proxy.channel.Channel;
import com.cetccloud.ap.proxy.channel.cossDomain.model.CallTypeEnum;
import com.cetccloud.ap.proxy.channel.cossDomain.model.CrossDomainModel;
import com.cetccloud.ap.proxy.processor.kafka.model.KafkaTransferParam;
import com.cetccloud.ap.util.IdUtil;
import com.cetccloud.ap.util.KafkaKeyUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * @ClassName KafkaExtractor
 * @Description TODO
 * @Author superz
 * @Date 2021/4/20 14:32
 * @Version 1.0
 **/
@Slf4j
@Component
public class KafkaTransferExtractor {

    @Autowired
    private ProxyConfig proxyConfig;

    @Autowired
    private Channel crossDomainChannel;

    /**
     * @KafkaListener(topicPattern = "topic.proxy.transfer.*", concurrency = "3")
     **/
    @KafkaListener(topics = "#{proxyConfig.getKafka.getTopicList()}", concurrency = "15")
    public void onMessage(ConsumerRecord<String, String> record) {
        log.info("接收kafka消息：KafkaExtractor [{}], message: [{}]", record.topic(), record.value());
        String key = record.key();
        if (StringUtils.isBlank(key)) {
            log.warn("消息体缺失destZoneId");
            return;
        }
        String destZoneId = KafkaKeyUtil.zoneByKey(key);

        String topicPrefix = proxyConfig.getKafka().getPrefix();
        KafkaTransferParam param = new KafkaTransferParam();
        param.setTopic(StringUtils.removeStart(record.topic(), topicPrefix));
        param.setMessage(record.value());
        // 发起跨域请求
        CrossDomainModel model = new CrossDomainModel();
        model.setTransId(IdUtil.transId());
        model.setCallType(CallTypeEnum.KAFKA_TRANSFER);
        model.setPayload(BeanUtil.beanToMap(param));
        crossDomainChannel.message(destZoneId, model);
    }
}
